package com.luoqifei.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.List;

public class HBaseTool {
    private static Logger log = LoggerFactory.getLogger(HBaseTool.class);
    private static Configuration hbaseConfig = null;

    public HBaseTool(String zkIp, String zkPort) {
        Configuration conf = new Configuration();
        //set zookeeper url
        conf.set("hbase.zookeeper.quorum", zkIp);
        conf.set("hbase.zookeeper.property.clientPort", zkPort);
        //conf.set("hbase.master","127.0.0.1:60000");
        hbaseConfig = HBaseConfiguration.create(conf);
    }

    public static void main(String[] args) {
        HBaseTool HBaseTool = new HBaseTool("10.1.10.38", "2181");
        HBaseTool.showAllRecords("test");
        //query by rowKey
        //HbaseService.showOneRecordByRowKey("test", "row1");
        //HbaseService.getRecordsByRowKeyAndColumn("test", "row1","cf");
        //HbaseService.getRecordsByRowKeyAndColumn("test", "row1","cf","a");

    }

    /**
     * create table
     *
     * @param tableName
     * @param columnFamilies
     * @throws Exception
     */
    public void createTable(String tableName, List<String> columnFamilies) throws Exception {
        log.info("start to create table : {}", tableName);
        HBaseAdmin hBaseAdmin = new HBaseAdmin(hbaseConfig);
        if (hBaseAdmin.tableExists(tableName)) {
            log.warn("Table {} had exists,we delete it.", tableName);
            hBaseAdmin.disableTable(tableName);
            hBaseAdmin.deleteTable(tableName);
            log.warn("Table {} delete successed.", tableName);
        }
        if (columnFamilies == null || columnFamilies.size() == 0) {
            log.error("the column family is null or size==0.");
            throw new Exception("when add table ,we need special the column family.");
        }
        HTableDescriptor tableDescriptor = new HTableDescriptor(tableName);
        for (String cf : columnFamilies) {
            tableDescriptor.addFamily(new HColumnDescriptor(cf));
        }
        hBaseAdmin.createTable(tableDescriptor);
        log.info("success to create table : {} .", tableName);
    }

    /**
     * drop table
     *
     * @param tableName
     * @return
     */
    public boolean dropTable(String tableName) {
        log.info("start to drop table {} .", tableName);
        try {
            HBaseAdmin hBaseAdmin = new HBaseAdmin(hbaseConfig);
            hBaseAdmin.disableTable(tableName);
            hBaseAdmin.deleteTable(tableName);
        } catch (IOException e) {
            log.error("fail to drop table {}", tableName);
            e.printStackTrace();
            return false;
        }
        log.info("success to drop table : {} .", tableName);
        return true;
    }

    /**
     * add column family in table
     */
    public void addColumn(String tableName, String columnName) {
        try {
            HBaseAdmin admin = new HBaseAdmin(hbaseConfig);
            HColumnDescriptor columnDescriptor = new HColumnDescriptor(columnName);
            //add column family
            admin.addColumn(tableName, columnDescriptor);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public void insertData(String tableName, String rowKey, String cf, HashMap<String, String> cValue) throws IOException {
        log.info("start to insert row data : table name is {},rowkey is {},column family is {},column and values " +
                "is {}.", tableName, rowKey, cf, cValue.toString());
        HTable table = new HTable(hbaseConfig, tableName);
        //add data to rowKey
        Put put = new Put(new String(rowKey).getBytes());
        //this data in cf column family,and in c column, the value
        for (String c : cValue.keySet()) {
            put.add(new String(cf).getBytes(), new String(c).getBytes(), new String(cValue.get(c)).getBytes());
        }
        table.put(put);
        table.flushCommits();
        log.info("success to insert data,row key is {}.", rowKey);
    }

    /**
     * add one row data in table
     *
     * @param tableName
     */

    public void showAllRecords(String tableName) {
        log.info("start==============show All Records=============");
        HTable table = null;
        try {
            table = new HTable(hbaseConfig, tableName);
        } catch (IOException e) {
            e.printStackTrace();
        }
        try {
            Scan scan = new Scan();
            ResultScanner rss = table.getScanner(scan);

            for (Result r : rss) {
                System.out.println("\n row: " + new String(r.getRow()));

                for (KeyValue kv : r.raw()) {

                    System.out.println("family=>" + new String(kv.getFamily(), "utf-8")
                            + "  value=>" + new String(kv.getValue(), "utf-8")
                            + "  qualifer=>" + new String(kv.getQualifier(), "utf-8")
                            + "  timestamp=>" + kv.getTimestamp());
                }
            }
            rss.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
        log.info("end==============show All Records=============");
    }

    public void showOneRecordByRowKey(String tableName, String rowkey) {
        HTable table = null;
        try {
            table = new HTable(hbaseConfig, tableName);
        } catch (IOException e) {
            e.printStackTrace();
        }

        try {
            Get get = new Get(rowkey.getBytes());
            Result r = table.get(get);
            System.out.println("start===showOneRecordByRowKey==row: " + "\n");
            System.out.println("row: " + new String(r.getRow(), "utf-8"));

            for (KeyValue kv : r.raw()) {
                String timestampFormat = new SimpleDateFormat("yyyy-MM-dd HH:MM:ss").format(new Date(kv.getTimestamp()));
                System.out.println("\nKeyValue: " + kv);
                System.out.println("key: " + kv.getKeyString());
                System.out.println("family=>" + new String(kv.getFamily(), "utf-8")
                        + "  value=>" + new String(kv.getValue(), "utf-8")
                        + "  qualifer=>" + new String(kv.getQualifier(), "utf-8")
                        + "  timestamp=>" + timestampFormat);

            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        log.info("end===========showOneRecordByRowKey");
    }

    public Result getRecordsByRowKeyAndColumn(String tableName, String rowKey, String column) throws IOException {
        log.info("start to get records by rowKey and column.");
        HTable table = new HTable(hbaseConfig, tableName);
        Get get = new Get(rowKey.getBytes());
        get.addFamily(column.getBytes());
        log.info("success to get record by rowKey {},and column {} .", rowKey, column);
        return table.get(get);
    }

    public Result getOneRecordByRowKeyAndQualifier(String tableName, String rowKey, String column, String qualifier) throws IOException {
        log.info("start to get one record by rowKey ,column and qualifier. ");
        HTable table = new HTable(hbaseConfig, tableName);
        Get get = new Get(rowKey.getBytes());
        get.addColumn(Bytes.toBytes(column), Bytes.toBytes(qualifier));
        Result r = table.get(get);
        log.info("success to get record by rowKey {},and qualifier {} .", rowKey, qualifier);
        return r;
    }

    public String GetTimeByStamp(String timestamp) {
        long datatime = Long.parseLong(timestamp);
        Date date = new Date(datatime);
        SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:MM:ss");
        String timeresult = format.format(date);
        System.out.println("Time : " + timeresult);
        return timeresult;

    }

    public String GetStampByTime(String time) {
        String Stamp = "";
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        Date date;
        try {
            date = sdf.parse(time);
            Stamp = date.getTime() + "000";
            System.out.println(Stamp);

        } catch (Exception e) {
            e.printStackTrace();
        }
        return Stamp;
    }

}

